package org.gbif.metrics.cli;

import org.gbif.api.model.occurrence.Occurrence;
import org.gbif.common.messaging.api.MessageCallback;
import org.gbif.common.messaging.api.OccurrencePersistedMessage;
import org.gbif.metrics.cube.HBaseCubes;
import org.gbif.metrics.cube.tile.density.DensityCube;
import org.gbif.metrics.cube.tile.density.DensityCubeUtil;
import org.gbif.metrics.cube.tile.density.DensityCubeUtil.Op;
import org.gbif.metrics.cube.tile.density.DensityTile;

import java.io.IOException;

import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.urbanairship.datacube.Batch;
import org.apache.hadoop.conf.Configuration;

/**
 * The service to update density cubes.
 */
class DensityCubeUpdaterService extends CubeUpdaterService<DensityCubeConfiguration> {

  DensityCubeUpdaterService(DensityCubeConfiguration configuration) {
    super(configuration);
  }

  @Override
  MessageCallback<OccurrencePersistedMessage> getCallback(final DensityCubeConfiguration configuration,
    final Configuration hadoopConfiguration) {

    return new CubeMutatorCallback<DensityTile>() {

      private volatile HBaseCubes<DensityTile> singletonCube;

      @Override
      protected synchronized HBaseCubes<DensityTile> getCube() {
        if (singletonCube == null) {
          Preconditions.checkNotNull(configuration.cubeTable); // should have been caught before, but for safety
          Preconditions.checkNotNull(configuration.columnFamily); // should have been caught before, but for safety
          Preconditions.checkNotNull(configuration.cubeCounterTable);
          Preconditions.checkNotNull(configuration.cubeLookupTable);
          try {
            singletonCube =
              HBaseCubes.newCombiningBatchAsync(DensityCube.INSTANCE, DensityTile.DESERIALIZER,
                configuration.cubeTable.getBytes(), configuration.cubeCounterTable.getBytes(),
                configuration.cubeLookupTable.getBytes(), configuration.columnFamily.getBytes(), hadoopConfiguration,
                configuration.writeBatchSize);

          } catch (IOException e) {
            throw Throwables.propagate(e);
          }
        }
        return singletonCube;
      }

      @Override
      protected Batch<DensityTile> getNewMutations(Occurrence occurrence) {
        return DensityCubeUtil.cubeMutations(occurrence, Op.ADDITION, configuration.numberZooms,
          configuration.pixelsPerCluster);
      }

      @Override
      protected Batch<DensityTile> getUpdateMutations(Occurrence oldOccurrence, Occurrence newOccurrence) {
        Batch<DensityTile> update =
          DensityCubeUtil.cubeMutations(oldOccurrence, Op.SUBTRACTION, configuration.numberZooms,
            configuration.pixelsPerCluster);
        if (newOccurrence != null) { // a delete has no new occurrence
          update.putAll(DensityCubeUtil.cubeMutations(newOccurrence, Op.ADDITION, configuration.numberZooms,
            configuration.pixelsPerCluster));
        }
        return update;
      }
    };
  }
}
